package cn.crsc.openlookeng.adhoc.query;

import cn.crsc.openlookeng.adhoc.dataframe.ColumnInfo;
import cn.crsc.openlookeng.adhoc.dataframe.PagedDataFrame;
import cn.crsc.openlookeng.adhoc.dataframe.Row;
import cn.crsc.openlookeng.adhoc.jdbc.JdbcDao;
import cn.crsc.openlookeng.adhoc.jdbc.QueryObject;
import cn.crsc.openlookeng.config.DynamicDataSourceRouting;
import cn.crsc.openlookeng.domain.Page;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.convert.ConversionService;
import org.springframework.format.support.DefaultFormattingConversionService;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.stream.Collectors;

@Component
@Slf4j
public class QueryServiceImpl implements QueryService, SmartInitializingSingleton {


    @Autowired
    DynamicDataSourceRouting dynamicDataSourceRouting;

    @Autowired(required = false)
    protected ConversionService conversionService;

    public QueryServiceImpl(){

    }

    @Override
    public Map<String, Object> executeQuery(QueryObject queryObject) throws Exception {
        return getData(executeDataFrame(queryObject), false);
    }

    private Map<String, Object> getData(PagedDataFrame executeDataFrame, boolean convert) {
        Map<String, Object> result = new HashMap<>();

        List<String> metaData = new ArrayList<>();
        for (ColumnInfo columnInfo : executeDataFrame.getNowDataFrame().getMetaData().getColumns()) {
            metaData.add(columnInfo.getName());
        }

        result.put("schema", metaData);

        //如果convert为true  按列返回信息
        if (convert) {
            for (String colName : metaData) {
                result.put(colName, new ArrayList<Object>());
            }

            for (Row row : executeDataFrame.getNowDataFrame()) {
                for (String colName : metaData) {
                    ((List<Object>) result.get(colName)).add(row.getValueAs(colName));
                }
            }
        } else {
            //convert为false 按行返回信息
            List<Map<String, Object>> data = new ArrayList<>();
            for (Row row : executeDataFrame.getNowDataFrame()) {
                data.add(row.getRowMap());
            }
            result.put("rows", data);
        }

        return result;
    }

    @Override
    public PagedDataFrame executeDataFrame(QueryObject queryObject) throws Exception {

        PagedDataFrame dataFrame = null;
        int retry = 0;
        while (retry < 3) {
            log.debug(queryObject.getSql());
            Connection connection = getConnection(queryObject);
            if(null != queryObject.getEngineType()){
                connection.setCatalog(queryObject.getEngineType().getDriver());
            }
            if(null != queryObject.getSchema()){
                connection.setSchema(queryObject.getSchema());
            }
            try {
                dataFrame = JdbcDao.queryAsDataFrame(queryObject, connection);
                break;
            } catch (RuntimeException e) {
                e.printStackTrace();
                log.error(e.getMessage());
                retry += 1;
                Thread.sleep(1000);
            }
        }

        return dataFrame;
    }

    @Override
    public List<String> getSchemas(QueryObject queryObject) throws Exception {

        Connection connection = getConnection(queryObject);
        ResultSet resultSet = JdbcDao.query(queryObject, connection);
        List<String> schemas = new ArrayList<>();
        while (resultSet.next()) {
            schemas.add(resultSet.getString(1));
        }
        return schemas;
    }

    @Override
    public List<String> getTables(QueryObject queryObject) throws SQLException {
        Connection connection = getConnection(queryObject);
        ResultSet resultSet = JdbcDao.query(queryObject, connection);
        List<String> tables = new ArrayList<>();
        while (resultSet.next()) {
            tables.add(resultSet.getString(1));
        }
        return tables;
    }

    @Override
    public Map<String, String> getTableInfo(QueryObject queryObject) throws SQLException {
        Connection connection = getConnection(queryObject);
        ResultSet resultSet = JdbcDao.query(queryObject, connection);
        Map<String, String> tableInfo = new HashMap<>();
        while (resultSet.next()) {
            tableInfo.put(resultSet.getString(1), resultSet.getString(2));
        }
        return tableInfo;
    }

    @Override
    public Object executeQueryForEntity(QueryObject queryObject, Class returnType, Class parameterizedType) {

        try {

            PagedDataFrame pagedDataFrame = executeDataFrame(queryObject);
            Map<String, Object> mapData = getData(pagedDataFrame, false);
            List<Map<String, Object>> data = (List<Map<String, Object>>) mapData.get("rows");
            String jsonData = "";

            if(List.class.isAssignableFrom(returnType)){

                if(null == parameterizedType){
                    throw new RuntimeException("请使用泛型参数转换：sql == " + queryObject.getSql());
                }

                // List<Map<>>
                if(Map.class.isAssignableFrom(parameterizedType)){
                    return data;
                }

                // List<Integet>
                if(BeanUtils.isSimpleValueType(parameterizedType)){
                    ArrayList<@Nullable Object> objects = Lists.newArrayList();
                    for (Map<String, Object> datum : data) {
                        datum.values().stream().forEach(
                                value->{
                                    Object obj = convertSimpleValue(value, parameterizedType);
                                    objects.add(obj);
                                }
                        );

                    }
                    return objects;

                }

                // List<User>
                jsonData = JsonUtil.obj2String(data);
                return JsonUtil.string2Obj(jsonData, List.class, parameterizedType);
            }

            if(Map.class.isAssignableFrom(returnType) && data.size() ==1){
                return data.get(0);
            }

            //Page<User>
            if(Page.class.isAssignableFrom(returnType)){
                if(null == parameterizedType){
                    throw new RuntimeException("请使用泛型参数转换：sql == " + queryObject.getSql());
                }
                jsonData = JsonUtil.obj2String(data);
                List content = JsonUtil.string2Obj(jsonData, List.class, parameterizedType);
                Page page = new Page(content,pagedDataFrame.getTotalCount());
                return page;
            }

            // User
            if(null == parameterizedType){
                //Integer
                if(BeanUtils.isSimpleValueType(returnType)){
                    Object source = data.get(0).values().stream().collect(Collectors.toList()).get(0);
                    return convertSimpleValue(source,returnType);
                }
                // User
                jsonData = JsonUtil.obj2String(data.get(0));
                return JsonUtil.string2Obj(jsonData,returnType);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;

    }

    private Connection getConnection(QueryObject queryObject) throws SQLException{
        DataSource resolvedDefaultDataSource = dynamicDataSourceRouting.getResolvedDefaultDataSource();
        Connection connection = resolvedDefaultDataSource.getConnection();
        if(null != queryObject.getEngineType()){
            connection.setCatalog(queryObject.getEngineType().getDriver());
        }
        if(null != queryObject.getSchema()){
            connection.setSchema(queryObject.getSchema());
        }
        return connection;
    }

    private Object convertSimpleValue(Object source,Class target){
        if(target.isAssignableFrom(source.getClass())){
            return source;
        }
        if(this.conversionService.canConvert(source.getClass(),target)){
            return this.conversionService.convert(source, target);
        }else {
            throw new RuntimeException("can not convert from "+source.getClass() +" to "+target);
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        if(null == this.conversionService){
            this.conversionService = new DefaultFormattingConversionService();
        }
    }
}
